package com.ds.lens.flink.monitor.forecast.function;

import com.google.gson.Gson;
import com.ds.lens.data.common.document.ForecastKafkaData;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.forecast.ForecastConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.nio.charset.StandardCharsets;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-08
 */
@Slf4j
public class ForecastKafkaDataSinkFunction implements ElasticsearchSinkFunction<ForecastKafkaData> {

    private static final long serialVersionUID = 721177339241185876L;

    private IndexRequest createIndexRequest(ForecastKafkaData element) {
        try {
            ForecastConfig forecastConfig = MonitorJobContext.getInstance().getForecastConfig();
            Gson gson = MonitorJobContext.getInstance().getGson();
            return Requests.indexRequest()
                    .index(forecastConfig.getEsForecastKafkaDataIndex())
                    .type(forecastConfig.getEsForecastKafkaDataType())
                    .source(gson.toJson(element).getBytes(StandardCharsets.UTF_8), XContentType.JSON);
        } catch (Exception e) {
            log.error("ForecastKafkaDataSinkFunction sink failed.", e);
            return null;
        }
    }

    @Override
    public void process(ForecastKafkaData forecastKafkaData, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
        IndexRequest request = createIndexRequest(forecastKafkaData);
        if (request != null) {
            requestIndexer.add(request);
        }
    }
}
